package org.apache.hadoop.hdfs.qjournal.client;

import java.io.IOException;

import org.apache.hadoop.hdfs.server.namenode.EditLogOutputStream;
import org.apache.hadoop.hdfs.server.namenode.EditsDoubleBuffer;
import org.apache.hadoop.hdfs.server.namenode.FSEditLogOp;
import org.apache.hadoop.io.DataOutputBuffer;

/**
 * EditLogOutputStream implementation that writes to a quorum of
 * remote journals.
 */
class QuorumOutputStream extends EditLogOutputStream {
  /** 使用AsyncLoggerSet来负责将新的editlog发送到远程JournalNode */
  private final AsyncLoggerSet loggers;
  /** 双缓存技术，可以让一个缓存的数据被flush out的同时，另外一个缓存的数据正在被写入，然后交换角色。 */
  private EditsDoubleBuffer buf;
  /** 这是整个segment文件的Id，在FSEditLog.startLogSegment()中被设置 */
  private final long segmentTxId;
  private final int writeTimeoutMs;

  public QuorumOutputStream(AsyncLoggerSet loggers, long txId, int outputBufferCapacity, int writeTimeoutMs) throws IOException {
    super();
    // 不同的流，对应的双缓冲都是自己独立的，

    this.buf = new EditsDoubleBuffer(outputBufferCapacity);
    this.loggers = loggers;
    this.segmentTxId = txId;
    this.writeTimeoutMs = writeTimeoutMs;
  }

  @Override
  public void write(FSEditLogOp op) throws IOException {
    buf.writeOp(op);
  }

  @Override
  public void writeRaw(byte[] bytes, int offset, int length) throws IOException {
    buf.writeRaw(bytes, offset, length);
  }

  @Override
  public void create(int layoutVersion) throws IOException {
    throw new UnsupportedOperationException();
  }

  @Override
  public void close() throws IOException {
    if (buf != null) {
      buf.close();
      buf = null;
    }
  }

  @Override
  public void abort() throws IOException {
    QuorumJournalManager.LOG.warn("Aborting " + this);
    buf = null;
    close();
  }

  @Override
  public void setReadyToFlush() throws IOException {
    buf.setReadyToFlush();
  }

  /**
   * flushAndSync()负责将缓存的待发送的数据拷贝一份出来，然后调用AsyncLoggerSet.sendEdits()，将数据发送给远程JournalNode，
   * 注意，这个发送也不是同步的，而是一个类似Future的操作，返回了一个获取结果的句柄，然后通过waitForWriteQuorum（）来同步等待响应，直到确认数据发送成功。
   *
   * 需要先异步发送，再同步等待结果，这是因为远程 接收数据的RPC服务器是由多台JournalNode组成的Quorum，因此需要并行发送以提高吞吐量。
   * AsyncLoggerSet封装和管理了对远程Quorum集群的操作。
   *
   * QuorumOutputStream.flushAndSync()通过创建调用请求，将这些请求逐个发送给所有的QuorumNode，
   * 然后，通过调用QuorumCall.create()，将句柄交给通过QuorumCall进行代理，
   * 然后，调用QuorumCall.waitFor()，等待响应。
   */
  @Override
  protected void flushAndSync(boolean durable) throws IOException {
    int numReadyBytes = buf.countReadyBytes(); // 已经处于ready状态的buf的大小
    if (numReadyBytes > 0) {
      int numReadyTxns = buf.countReadyTxns(); // readyBuffer中的操作数量大小
      long firstTxToFlush = buf.getFirstReadyTxId(); // 这个Buffer中第一个OP的transactionId

      assert numReadyTxns > 0;

      // 将双缓存内存中的数据 拷贝到一份新的字节数组中去
      // 1. IPC代码，rpc调用，它需要一次性将缓冲区里的数据全部发送到journalnode上去，通过rpc接口请求没有办法每次就发送一个大的数组的一部分的。
      // 2. 因为对journalnode的调用是异步的，他需要一个防御性的拷贝，来避免在发送这段数据之前，导致这个buffer缓冲被人修改了

      // 它的意思是什么？
      // 需要一次性的把所有的buffer里的数据发送到journal node里去，因为它是异步的，为了避免在异步发送的时候，还没发送，buffer的数据被人给修改了，
      // 所以在这里，它会将buffer里的数据先拷贝到一个新的字节数组里去
      DataOutputBuffer bufToSend = new DataOutputBuffer(numReadyBytes);
      // 将缓存在内存中的数据放入DataOutputBuffer,准备发送，实际上将数据拷贝了一份出来
      buf.flushTo(bufToSend);
      // 上面两行代码，其实就是将buffer缓冲里的数据，先写入一个新的DataOutputBuffer里去，
      assert bufToSend.getLength() == numReadyBytes;
      // 这里就是将新的DataOutputBuffer里的数据放到一个byte[] data数组里去
      byte[] data = bufToSend.getData();
      assert data.length == bufToSend.getLength();

      // 通过AsyncLogger，将这些edits log发送到大所述的journal node上去，异步方式，返回一个获取结果的类似Future的句柄
      // 每个AsyncLogger 都是基于底层的一个单线程池异步发送网络请求，rpc接口调用，
      // QuorumCall里可以获取到每个AsyncLogger对应的Future
      QuorumCall<AsyncLogger, Void> qcall = loggers.sendEdits( segmentTxId, firstTxToFlush, numReadyTxns, data);
      // 同步等待返回结果，由于是paxos协议，因此只需要等待大多数节点返回成功，就可以认为成功返回
      // 在这里它必须等待写edits log到大多数journalnode都成功以后，才可以继续往下运行
      // 在这里会阻塞住，等待
      // 通过这里，观察每个Futrure结果，来实现quorum算法
      // 3个journalnode，必须有 (3/2)+1 = 2个都推送成功了才可以
      // 5个journalnode，必须有 (5/2)+1 = 3个都推送成功了才可以
      loggers.waitForWriteQuorum(qcall, writeTimeoutMs, "sendEdits"); // TODO
      
      // 因为我们成功地写了这批, 让 loggers 知道. 任何 future RPCs 都会让 loggers 知道最近的事务,  即使一个logger落后了。
      loggers.setCommittedTxId(firstTxToFlush + numReadyTxns - 1);
    }
  }

  @Override
  public String generateReport() {
    StringBuilder sb = new StringBuilder();
    sb.append("Writing segment beginning at txid " + segmentTxId + ". \n");
    loggers.appendReport(sb);
    return sb.toString();
  }
  
  @Override
  public String toString() {
    return "QuorumOutputStream starting at txid " + segmentTxId;
  }
}
